package com.cloudputing.mapreduce.datajoin;

import org.apache.hadoop.contrib.utils.join.DataJoinReducerBase;
import org.apache.hadoop.contrib.utils.join.TaggedMapOutput;
import org.apache.hadoop.io.Text;

public class DataJoinReducer extends DataJoinReducerBase {
	protected TaggedMapOutput combine(Object[] tags, Object[] values) {
		if (tags.length < 2)
			return null;
		String joinedStr = "";
		for (int i = 0; i < values.length; i++) {
			if (i > 0)
				joinedStr += ",";
			DataJoinTaggedWritable tw = (DataJoinTaggedWritable) values[i];
			String line = ((Text) tw.getData()).toString();
			String[] tokens = line.split(",", 2);
			joinedStr += tokens[1];
		}
		DataJoinTaggedWritable retv = new DataJoinTaggedWritable(new Text(
				joinedStr));
		retv.setTag((Text) tags[0]);
		return retv;
	}
}